package net.bwie.realtime.jtp.dws.trade.job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.dws.trade.bean.LiveRoomEventBean;
import net.bwie.realtime.jtp.dws.trade.function.AggregateLiveRoomEventMetrics;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import com.alibaba.fastjson.JSON;

import java.time.Duration;
import java.time.format.DateTimeFormatter;
import java.time.LocalDateTime;
import java.time.ZoneId;

public class DwsLiveRoomEventJob {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(30000L);

        DataStream<String> eventLogDS = KafkaUtil.consumerKafka(env, "dwd_live_event_log");
        DataStream<String> newFansDS = KafkaUtil.consumerKafka(env, "dwd_user_new_fans_detail");
        DataStream<String> newFansTeamDS = KafkaUtil.consumerKafka(env, "dwd_user_new_fans_team_detail");
        DataStream<String> orderDS = KafkaUtil.consumerKafka(env, "dwd_trade_order_detail");
        DataStream<String> payDS = KafkaUtil.consumerKafka(env, "dwd_trade_pay_detail");

        DataStream<String> resultStream = handle(eventLogDS, newFansDS, newFansTeamDS, orderDS, payDS);
        resultStream.print("窗口聚合结果========>");

        // 先转成 JSON 字符串流
        SingleOutputStreamOperator<String> jsonStream = resultStream.map(JSON::toJSONString);

// 再调用工具类写入 Kafka
        KafkaUtil.producerKafka(jsonStream, "dws_live_room_event_metrics");

        env.execute("DwsLiveRoomEventJob");
    }

    private static SingleOutputStreamOperator<String> handle(
            DataStream<String> eventLogDS,
            DataStream<String> newFansDS,
            DataStream<String> newFansTeamDS,
            DataStream<String> orderDS,
            DataStream<String> payDS) {

        // ========= 事件日志 =========
        SingleOutputStreamOperator<LiveRoomEventBean> eventLogBeanDS = eventLogDS
                .map(s -> {
                    JSONObject j = JSON.parseObject(s);
                    LiveRoomEventBean b = new LiveRoomEventBean();

                    b.setLiveRoomId(j.getString("live_session_id"));
                    b.setAnchorId(String.valueOf(j.getLong("anchor_id")));
                    b.setUserId(j.getString("user_id"));

                    String type = j.getString("event_type");

                    // ✅ 统一 eventType 字段
                    if ("enter_live_room".equals(type) || "进入".equals(type)) {
                        b.setEventType("进入直播间");
                    } else if ("comment".equals(type) || "评论".equals(type)) {
                        b.setEventType("评论");
                    } else if ("click".equals(type) || "点击".equals(type)) {
                        b.setEventType("点击");
                    } else {
                        b.setEventType(type);
                    }

                    // ✅ 评论标记
                    if ("评论".equals(b.getEventType())) {
                        b.setIsNewComment(1);
                    }

                    // ✅ 停留时长（直接存字段，聚合器才能累加）
                    if (j.getLong("duration_sec") != null) {
                        b.setDurationSec(j.getLong("duration_sec"));
                    }

                    // ✅ 保留 enter_time/leave_time，聚合器可能会用
                    b.setEnterTime(j.getString("enter_time"));
                    b.setLeaveTime(j.getString("leave_time"));

                    // ✅ 用 enter_time 作为 ts
                    b.setTs(parseToMillisSafe(j.getString("enter_time")));
                    return b;
                })
                .returns(LiveRoomEventBean.class);

        // ========= 新增粉丝 =========
        SingleOutputStreamOperator<LiveRoomEventBean> newFansBeanDS = newFansDS
                .map(s -> {
                    JSONObject j = JSON.parseObject(s);
                    LiveRoomEventBean b = new LiveRoomEventBean();
                    b.setLiveRoomId(j.getString("live_session_id"));
                    b.setAnchorId(String.valueOf(j.getLong("anchor_id")));
                    b.setUserId(j.getString("user_id"));
                    b.setIsNew(j.getInteger("is_new"));
                    b.setIsFansTeam(j.getInteger("is_not_fan") != null && j.getInteger("is_not_fan") == 0 ? 1 : 0);
                    b.setEventType("关注");
                    b.setTs(parseToMillisSafe(j.getString("change_time")));
                    return b;
                })
                .returns(LiveRoomEventBean.class);

        // ========= 新增粉丝团 =========
        SingleOutputStreamOperator<LiveRoomEventBean> newFansTeamBeanDS = newFansTeamDS
                .map(s -> {
                    JSONObject j = JSON.parseObject(s);
                    LiveRoomEventBean b = new LiveRoomEventBean();
                    b.setLiveRoomId(j.getString("live_session_id"));
                    b.setAnchorId(j.getString("anchor_id"));
                    b.setUserId(j.getString("user_id"));
                    b.setEventType("加入粉丝团"); // ✅ 和聚合器统一
                    b.setTs(parseToMillisSafe(j.getString("event_time")));
                    return b;
                })
                .returns(LiveRoomEventBean.class);

        // ========= 下单 =========
        SingleOutputStreamOperator<LiveRoomEventBean> orderBeanDS = orderDS
                .map(s -> {
                    JSONObject j = JSON.parseObject(s);
                    LiveRoomEventBean b = new LiveRoomEventBean();
                    b.setLiveRoomId(j.getString("live_session_id"));
                    b.setAnchorId(String.valueOf(j.getLong("anchor_id")));
                    b.setUserId(j.getString("user_id"));
                    b.setEventType("下单");
                    b.setOrderAmount(j.getDouble("order_amount"));
                    b.setTs(parseToMillisSafe(j.getString("order_time")));
                    return b;
                })
                .returns(LiveRoomEventBean.class);

        // ========= 支付 =========
        SingleOutputStreamOperator<LiveRoomEventBean> payBeanDS = payDS
                .map(s -> {
                    JSONObject j = JSON.parseObject(s);
                    LiveRoomEventBean b = new LiveRoomEventBean();
                    b.setLiveRoomId(j.getString("live_session_id"));
                    b.setAnchorId(String.valueOf(j.getLong("anchor_id")));
                    b.setUserId(j.getString("user_id"));
                    b.setEventType("支付");
                    b.setOrderAmount(j.getDouble("pay_amount")); // ⚠️ 确认字段名
                    b.setTs(parseToMillisSafe(j.getString("pay_time")));
                    return b;
                })
                .returns(LiveRoomEventBean.class);

        // ========= 合并流 =========
        DataStream<LiveRoomEventBean> unionStream = eventLogBeanDS
                .union(newFansBeanDS, newFansTeamBeanDS, orderBeanDS, payBeanDS);

        // ========= 水位线 =========
        SingleOutputStreamOperator<LiveRoomEventBean> watermarkStream = unionStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<LiveRoomEventBean>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((SerializableTimestampAssigner<LiveRoomEventBean>) (element, recordTimestamp) -> {
                                    return element.getTs() == null ? 0L : element.getTs();
                                })
                );

        // ========= 窗口聚合 =========
        WindowedStream<LiveRoomEventBean, String, TimeWindow> windowedStream = watermarkStream
                .keyBy(bean -> bean.getLiveRoomId() + "_" + bean.getAnchorId())
                .window(TumblingEventTimeWindows.of(Time.seconds(30)));

        SingleOutputStreamOperator<LiveRoomEventBean> resultStream =
                windowedStream.apply(new AggregateLiveRoomEventMetrics());


        return resultStream.map(bean -> {
            return String.format(
                    "{" +
                            "\"window_start_time\":\"%s\"," +
                            "\"window_end_time\":\"%s\"," +
                            "\"event_minute\":\"%s\"," +
                            "\"live_room_id\":\"%s\"," +
                            "\"anchor_id\":\"%s\"," +
                            "\"deal_count\":%d," +
                            "\"deal_user_count\":%d," +
                            "\"deal_amount\":%.2f," +
                            "\"uv_count\":%d," +
                            "\"duration_sec\":%d," +
                            "\"new_fans_count\":%d," +
                            "\"new_fans_team_count\":%d," +
                            "\"comment_count\":%d," +
                            "\"ts\":%d" +
                            "}",
                    bean.getWindowStartTime() == null ? "" : bean.getWindowStartTime(),
                    bean.getWindowEndTime() == null ? "" : bean.getWindowEndTime(),
                    bean.getEventMinute() == null ? "" : bean.getEventMinute(),
                    bean.getLiveRoomId() == null ? "" : bean.getLiveRoomId(),
                    bean.getAnchorId() == null ? "" : bean.getAnchorId(),
                    bean.getDealCount() == null ? 0 : bean.getDealCount(),
                    bean.getDealUserCount() == null ? 0 : bean.getDealUserCount(),
                    bean.getDealAmount() == null ? 0.0 : bean.getDealAmount(),
                    bean.getUvCount() == null ? 0 : bean.getUvCount(),
                    bean.getDurationSec() == null ? 0 : bean.getDurationSec(),
                    bean.getNewFansCount() == null ? 0 : bean.getNewFansCount(),
                    bean.getNewFansTeamCount() == null ? 0 : bean.getNewFansTeamCount(),
                    bean.getCommentCount() == null ? 0 : bean.getCommentCount(),
                    bean.getTs() == null ? 0 : bean.getTs()
            );
        });

    }

    private static long parseToMillisSafe(String timeStr) {
        if (timeStr == null || timeStr.trim().isEmpty()) {
            return 0L;
        }
        try {
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            LocalDateTime ldt = LocalDateTime.parse(timeStr, formatter);
            return ldt.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
        } catch (Exception e) {
            return 0L; // ❌ 不用 System.currentTimeMillis()，避免窗口乱飞
        }
    }
}
